/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.cst.lls.worker;

import cn.ac.iie.di.cst.lls.common.RuntimeEnv;
import cn.ac.iie.di.cst.lls.se.CstdxStartup;
import cn.ac.iie.di.cst.lls.util.DataInfo;
//import static cn.iie.ac.di.cst.mavetest.dc.Test.encrypt;
//import static cn.iie.ac.di.cst.lls.en.util.DataEncryption.encrypt;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import org.apache.log4j.Logger;

/**
 *
 * @author Carlisle
 */
public class RocketdxConsumer {

    private static org.apache.log4j.Logger logger = Logger.getLogger(RocketdxConsumer.class.getName());
    private static DefaultMQPushConsumer consumer = null;
    public static AtomicInteger a=new AtomicInteger(0);
            
//    static volatile boolean stopped;

    public static void consumeData(String namesvr, String topic) throws MQClientException {
        consumer = new DefaultMQPushConsumer(RuntimeEnv.getParam(RuntimeEnv.ProductConsumersname).toString());
        consumer.setNamesrvAddr(namesvr);
        consumer.setInstanceName((String) RuntimeEnv.getParam(RuntimeEnv.ProductConsumersname) + "-" + RuntimeEnv.getParam(RuntimeEnv.HostNameIp));
        consumer.setConsumeConcurrentlyMaxSpan(65000);
        consumer.setPullThresholdForQueue(8);
        consumer.setPullBatchSize(8);
        consumer.setConsumeThreadMin(Integer.valueOf(RuntimeEnv.getParam(RuntimeEnv.ThSize).toString()));
        consumer.setConsumeThreadMax(Integer.valueOf(RuntimeEnv.getParam(RuntimeEnv.ThSize).toString()));
        consumer.setConsumeMessageBatchMaxSize(4);
        consumer.setClientCallbackExecutorThreads(24);
        consumer.setPullInterval(0);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        String[] namesrv = topic.split("[|]");
        for (int i = 0; i < namesrv.length; i++) {
            consumer.subscribe(namesrv[i] + "_mq", "*");
            logger.info(namesrv[i] + "_mq  consumer init succeff");
        }
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String top=msg.getTopic();
                        byte[] body=msg.getBody();
                        CstdxStartup.ab.put(new DataInfo(top,body));
                        a.incrementAndGet();
                    } catch (Exception ex) {
                        logger.warn(ex);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            consumer.start();
        } catch (Exception ex) {
            logger.error("consumer start error" + ex);
        }
    }

    public static void stop() {
//        stopped = true;
        if (consumer != null) {
            logger.info("shutdown consumer ");
            consumer.shutdown();
        }
    }

}
